#include "ce_core.h"

static const int    init_tasknum = 1024;
static ce_atomic_t  taskid = 0;

static void *
ce_thread_run(void *data) {
    ce_thread_pool_t       *pool = (ce_thread_pool_t *)data;

    ce_list_node_t         *node;
    ce_thread_task_t       *task;
    sigset_t                set;
    // pthread_t               tid;

    // tid = pthread_self();
    // printf("thread %p run\n", tid);

    sigfillset(&set);

    sigdelset(&set, SIGILL);
    sigdelset(&set, SIGFPE);
    sigdelset(&set, SIGSEGV);
    sigdelset(&set, SIGBUS);

    pthread_sigmask(SIG_BLOCK, &set, NULL);

    for (;;) {
        pthread_mutex_lock(&pool->mtx);

        while (ce_list_empty(&pool->task_queue)) {
            pthread_cond_wait(&pool->cond, &pool->mtx);
        }

        node = ce_list_pop_head(&pool->task_queue);
        pthread_mutex_unlock(&pool->mtx);

        task = ce_entry(node, ce_thread_task_t, qn);

        if (task->handler != NULL) {
            task->handler(task->ctx);
        }

        ce_pool_put(pool->task_pool, task);
    }

    return NULL;
}

ce_thread_pool_t *
ce_thread_pool_create(uint32_t num) {
    ce_thread_pool_t       *pool;
    pthread_t               tid;

    pool = malloc(sizeof(ce_thread_pool_t));

    pool->task_pool = ce_create_pool(sizeof(ce_thread_task_t), init_tasknum);
    ce_list_init(&pool->task_queue);

    // create threads
    pool->threadnum = num;

    pthread_mutex_init(&pool->mtx, NULL);
    pthread_cond_init(&pool->cond, NULL);

    for (int i = 0; i < num; ++i) {
        pthread_create(&tid, NULL, ce_thread_run, pool);
        pthread_detach(tid); 
    }

    return pool;
}

ce_thread_task_t *
ce_thread_task_get(ce_thread_pool_t *pool) {
    ce_thread_task_t       *task;

    task = ce_pool_get(pool->task_pool);

    task->ctx = NULL;
    task->handler = NULL;

    task->id = ce_atomic_fetch_add(&taskid, 1);

    return task;
}

void
ce_thread_task_post(ce_thread_pool_t *pool, ce_thread_task_t *task) {
    pthread_mutex_lock(&pool->mtx);

    ce_list_push_tail(&pool->task_queue, &task->qn);

    pthread_cond_signal(&pool->cond);

    pthread_mutex_unlock(&pool->mtx);
}

void
ce_routine_start(ce_thread_pool_t *pool,
    void (*handler)(void *), void *ctx) {

    ce_thread_task_t       *task;

    task = ce_thread_task_get(pool);
    task->ctx = ctx;
    task->handler = handler;

    ce_thread_task_post(pool, task);
}
